/*=========================================================================
 * Copyright (c) 2010-2011 VMware, Inc. All rights reserved.
 * This product is protected by U.S. and international copyright
 * and intellectual property laws. VMware products are covered by
 * one or more patents listed at http://www.vmware.com/go/patents.
 *=========================================================================
 */
package com.gemfire.onlineretailer.dao;


import static com.gemstone.gemfire.cache.RegionShortcut.PARTITION_REDUNDANT;
import static com.gemstone.gemfire.cache.RegionShortcut.REPLICATE;
import static com.gemstone.gemfire.cache.client.ClientRegionShortcut.CACHING_PROXY;
import static com.gemstone.gemfire.cache.client.ClientRegionShortcut.PROXY;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import com.gemfire.onlineretailer.dao.functions.AccountsPayableItemFunctions;
import com.gemfire.onlineretailer.dao.functions.AccountsPayableItemsCollector;
import com.gemfire.onlineretailer.domain.AccountHeads;
import com.gemfire.onlineretailer.domain.AccountType;
import com.gemfire.onlineretailer.domain.Book;
import com.gemfire.onlineretailer.domain.Clothing;
import com.gemfire.onlineretailer.domain.Disk;
import com.gemfire.onlineretailer.domain.Electronics;
import com.gemfire.onlineretailer.domain.Furniture;
import com.gemfire.onlineretailer.domain.Item;
import com.gemfire.onlineretailer.domain.ItemType;
import com.gemfire.onlineretailer.domain.PaymentVoucher;
import com.gemfire.onlineretailer.domain.ShoppingCart;
import com.gemfire.onlineretailer.domain.Store;
import com.gemfire.onlineretailer.domain.Vendor;
import com.gemfire.onlineretailer.util.LoggingCacheListener;
import com.gemfire.onlineretailer.util.SimpleCqListener;
import com.gemstone.gemfire.cache.Cache;
import com.gemstone.gemfire.cache.CacheFactory;
import com.gemstone.gemfire.cache.Region;
import com.gemstone.gemfire.cache.client.ClientCache;
import com.gemstone.gemfire.cache.client.ClientCacheFactory;
import com.gemstone.gemfire.cache.execute.Execution;
import com.gemstone.gemfire.cache.execute.FunctionService;
import com.gemstone.gemfire.cache.execute.ResultCollector;
import com.gemstone.gemfire.cache.query.CqAttributes;
import com.gemstone.gemfire.cache.query.CqAttributesFactory;
import com.gemstone.gemfire.cache.query.CqClosedException;
import com.gemstone.gemfire.cache.query.CqException;
import com.gemstone.gemfire.cache.query.CqExistsException;
import com.gemstone.gemfire.cache.query.CqListener;
import com.gemstone.gemfire.cache.query.CqQuery;
import com.gemstone.gemfire.cache.query.FunctionDomainException;
import com.gemstone.gemfire.cache.query.NameResolutionException;
import com.gemstone.gemfire.cache.query.Query;
import com.gemstone.gemfire.cache.query.QueryInvalidException;
import com.gemstone.gemfire.cache.query.QueryInvocationTargetException;
import com.gemstone.gemfire.cache.query.QueryService;
import com.gemstone.gemfire.cache.query.RegionNotFoundException;
import com.gemstone.gemfire.cache.query.SelectResults;
import com.gemstone.gemfire.cache.query.TypeMismatchException;
import com.gemstone.gemfire.management.internal.cli.util.JsonUtil.Employee;

/**
 * Data Access Object to store PaymentVouchers using GemFire.  
 * Calling the {@link #initClient()} will
 * connect to the distributed system as a GemFire client.
 * 
 * Once one of the init method is called, the rest of the methods in this DAO
 * can be used to query and update the store.
 * 
 * @author GemStone Systems, Inc.
 * @author Binit Datta
 */
public class OnlineRetailerDAO {
  
  private Region<String,Employee> employees;
  private Region<String,Item> items;
  private Region<String,Vendor> vendors;
  private Region<String,AccountHeads> accountHeads;
  private Region<String,Store> stores;
  
  private Region<String,ItemType> itemTypes;
  private Region<String,AccountType> accountTypes;
  
  private Region<String,Book> books;
  private Region<String,Clothing> clothes;
  private Region<String,Disk> disks;
  private Region<String,Electronics> electronics;
  private Region<String,Furniture> furnitures;

  

  private Region<String,PaymentVoucher> paymentVouchers;
  
  private Region<String,ShoppingCart> shoppingCarts;

  
  private final BufferedReader stdinReader = new BufferedReader(new InputStreamReader(System.in));


  ClientCache cache;
  

  private boolean isClient;

  /**
   * Connect to the distributed system as a peer and declare
   * the people and posts regions.
   */
  @SuppressWarnings("unchecked")
  public void initPeer() {
    //Connect to the distributed system. We're connecting to a known locator, 
    //which GemFire will use to discover other peers.
    Cache cache = new CacheFactory()
      .set("locators", "localhost[55221]")
      .set("mcast-port", "0")
      .set("log-level", "error")
      .create();
    
    //Create a listener that will print to standard out
    LoggingCacheListener listener = new LoggingCacheListener();
    
    //Declare the people region. After this call completes,
    //this member may have copied the contents of the region
    //from a peer
    itemTypes = cache.<String, ItemType>createRegionFactory(REPLICATE)
      .addCacheListener(listener)
      .create("itemTypes");
    
 
    
    accountTypes = cache.<String, AccountType>createRegionFactory(REPLICATE)
    	      .addCacheListener(listener)
    	      .create("accountTypes");
    
    //Declare the posts region.
    employees =  cache.<String, Employee>createRegionFactory(PARTITION_REDUNDANT)
      .addCacheListener(listener)
      .create("employees");
    
    vendors =  cache.<String, Vendor>createRegionFactory(PARTITION_REDUNDANT)
    	      .addCacheListener(listener)
    	      .create("vendors");
    
    stores =  cache.<String, Store>createRegionFactory(PARTITION_REDUNDANT)
    	      .addCacheListener(listener)
    	      .create("stores");
    
    accountHeads =  cache.<String, AccountHeads>createRegionFactory(PARTITION_REDUNDANT)
    	      .addCacheListener(listener)
    	      .create("accountHeads");
    
 
  
    paymentVouchers =  cache.<String, PaymentVoucher>createRegionFactory(PARTITION_REDUNDANT)
    	      .addCacheListener(listener)
    	      .create("paymentVouchers");
    
    books = cache.<String, Book>createRegionFactory(PARTITION_REDUNDANT)
  	      .addCacheListener(listener)
  	      .create("books");
    
    clothes = cache.<String, Clothing>createRegionFactory(PARTITION_REDUNDANT)
  	      .addCacheListener(listener)
  	      .create("clothes");
    
    electronics = cache.<String, Electronics>createRegionFactory(PARTITION_REDUNDANT)
  	      .addCacheListener(listener)
  	      .create("electronics");
    
    disks = cache.<String, Disk>createRegionFactory(PARTITION_REDUNDANT)
  	      .addCacheListener(listener)
  	      .create("disks");
    
    furnitures = cache.<String, Furniture>createRegionFactory(PARTITION_REDUNDANT)
  	      .addCacheListener(listener)
  	      .create("furnitures");
    
  }
  
  /** Connect to the distributed system as a client and 
   * declare the people and posts regions
 * @throws CqException 
 * @throws CqExistsException 
 * @throws QueryInvalidException 
 * @throws RegionNotFoundException 
 * @throws CqClosedException 
   */

  public void initClient() throws CqClosedException, RegionNotFoundException, QueryInvalidException, CqExistsException, CqException {
    //Connect to the system as a client. We're using a known locator
    //to discover servers in the distributed system.
    //
    //In addition, this client can subscribe to updates
    //from the servers.
     cache = new ClientCacheFactory()
      .addPoolLocator("localhost", 55221)
      .setPoolSubscriptionEnabled(true)
      .setPoolSubscriptionRedundancy(1)
      .set("log-level", "error")
      .create();
    
    //Create a listener that will print to standard out
    LoggingCacheListener listener = new LoggingCacheListener();
    
    
    //Declare the people region.
    
    employees = cache.<String, Employee>createClientRegionFactory(PROXY)
    	      .create("employees");
    
  
    
    itemTypes = cache.<String, ItemType>createClientRegionFactory(PROXY)
  	      .create("itemTypes");
    
    accountTypes = cache.<String, AccountType>createClientRegionFactory(PROXY)
  	      .create("accountTypes");
    
    vendors = cache.<String, Vendor>createClientRegionFactory(CACHING_PROXY)
  	      .create("vendors");
    
    createVendors();
    
    stores = cache.<String, Store>createClientRegionFactory(PROXY)
  	      .create("stores");
    
    accountHeads = cache.<String, AccountHeads>createClientRegionFactory(PROXY)
  	      .create("accountHeads");
    
    createAccounts();
    
  
    
 
    
    items = cache.<String, Item>createClientRegionFactory(PROXY)
  	      .create("items");
  
    createItems();
    
  
    
    paymentVouchers = cache.<String, PaymentVoucher>createClientRegionFactory(PROXY)
  	      .create("paymentVouchers");
    
    books = cache.<String, Book>createClientRegionFactory(PROXY)
    	      .create("books");
    
    createBooks();
    
    clothes = cache.<String, Clothing>createClientRegionFactory(PROXY)
    	      .create("clothes");
    
    createClothes();
    
    electronics = cache.<String, Electronics>createClientRegionFactory(PROXY)
    	      .create("electronics");
    
    createElectronics();
    
    furnitures = cache.<String, Furniture>createClientRegionFactory(PROXY)
    	      .create("furnitures");
    
    createFurnitures();
    
    disks = cache.<String, Disk>createClientRegionFactory(PROXY)
    	      .create("disks");
    
   createDisks();
    
    //Subscribe to updates for the people region. We're
    //subscribing to all updates to this region so we can
    //cache the people locally.
    vendors.registerInterestRegex(".*");
    
    registerPaymentCQ();
    
    isClient = true;
  }
  
  public List executeItemFunction() throws Exception {
	   Region<String, Item> itemRegion = cache.getRegion("items");
	   
	   writeToStdout("Item region \"" + itemRegion.getFullPath()
		        + "\" created in cache.");
	   
	   writeToStdout("Executing Function : AccountsPayableItemFunctions on region \""
		        + itemRegion.getFullPath()
		        + "\" with filter size " + 3 + " and with AccountsPayableItemsCollector.");
	   
	   AccountsPayableItemFunctions apFunction = new AccountsPayableItemFunctions();
	   
	   FunctionService.registerFunction(apFunction);
	   
	   writeToStdout("Press Enter to continue.");
	    stdinReader.readLine();
	    
	    Set<String> itemKeysForGet = new HashSet<String>();
	    itemKeysForGet.add(Item.ITEM_1_CODE);
	    itemKeysForGet.add(Item.ITEM_2_CODE);
	    itemKeysForGet.add(Item.ITEM_3_CODE);
	    
	    Execution execution = FunctionService.onRegion(itemRegion).withFilter(
	    		itemKeysForGet).withArgs(Boolean.TRUE).withCollector(
	            new AccountsPayableItemsCollector());
	        ResultCollector rc = execution.execute(apFunction);
	        
	        writeToStdout("Function executed successfully. Now getting the result");

	        List result = (List)rc.getResult();
	        
	        return result;
	
  }

 

  /**
   * Add a new person
   * @param name The name of the person to add.
   * @param profile The profile for this person
   */
  public void addPaymentVoucher(String voucherNumber, PaymentVoucher voucher) {
	  paymentVouchers.put(voucherNumber, voucher);
  }
  
  
public void addShoppingCart(String orderNumber, ShoppingCart shoppingCart){
	  shoppingCarts.put(orderNumber, shoppingCart);

}
  /**
   * Get this list of people in the system
   */
  public Set<ItemType> getItemTypes() {
    Set<Map.Entry<String,ItemType>> itemTypesCache = itemTypes.entrySet();
    Set<ItemType> itemTypes = new HashSet<ItemType>();
    
    for (Map.Entry<String,ItemType> it : itemTypesCache){
    	itemTypes.add(it.getValue());
    }
    
    return itemTypes;
  }
  
  public Set<AccountType> getAccountTypes() {
	    Set<Map.Entry<String,AccountType>> accountTypesCache = accountTypes.entrySet();
	    Set<AccountType> accountTypes = new HashSet<AccountType>();
	    
	    for (Map.Entry<String,AccountType> it : accountTypesCache){
	    	accountTypes.add(it.getValue());
	    }
	    
	    return accountTypes;
	  }
  
 
 
  public Employee getEmployee(String employeeId){
	  Employee emp = null;
	  if (employees.containsKey(employeeId)){
		  emp =  (Employee) employees.get(employeeId);
	  }
	  
	  return emp;
  }
  
  public Vendor getVendor(String vendorId){
	  Vendor vendor = null;
	  if (vendors.containsKey(vendorId)){
		  vendor =  (Vendor) vendors.get(vendorId);
	  }
	  
	  return vendor;
  }
  
  public Store getStore(String storeId){
	  Store store = null;
	  if (stores.containsKey(storeId)){
		  store =  (Store) stores.get(storeId);
	  }
	  
	  return store;
  }
  
  public Item getItems(String itemId){
	  Item item = null;
	  if (items.containsKey(itemId)){
		  item =  (Item) items.get(itemId);
	  }
	  
	  return item;
  }
  
  public AccountHeads getAccountHead(String accountId){
	  System.out.println("Received Account ID: "+accountId);
	  AccountHeads account = null;
	  if (accountHeads.containsKey(accountId)){
		  System.out.println("inside the iff Received Account ID: "+accountId);

		  account =  (AccountHeads) accountHeads.get(accountId);
		  
	  }
	  
	  account =  (AccountHeads) accountHeads.get(accountId);

	  return account;
  }
  
  public Book getBook(String bookId){
	  System.out.println("Received Book ID: "+bookId);
	  Book book = null;
	  if (books.containsKey(bookId)){
		  System.out.println("inside the iff Received Book ID: "+bookId);

		  book =  (Book) books.get(bookId);
		  
	  }
	  
	  book =  (Book) books.get(bookId);
	  return book;
  }
  
  public Clothing getClothing(String clothingId){
	  System.out.println("Received Clothing ID: "+clothingId);
	  Clothing cloth = null;
	  if (clothes.containsKey(clothingId)){
		  System.out.println("inside the iff Received Clothing ID: "+clothingId);

		  cloth =  (Clothing) clothes.get(clothingId);
		  
	  }
	  
	  cloth =  (Clothing) clothes.get(clothingId);

	  return cloth;
  }
  
  public Disk getDisk(String diskId){
	  System.out.println("Received Account ID: "+diskId);
	  Disk disk = null;
	  if (accountHeads.containsKey(diskId)){
		  System.out.println("inside the iff Received Disk ID: "+diskId);

		  disk =  (Disk) disks.get(diskId);
		  
	  }
	  
	  disk =  (Disk) disks.get(diskId);

	  return disk;
  }
  
  public Furniture getFurniture(String furnitureId){
	  System.out.println("Received furniture ID: "+furnitureId);
	  Furniture account = null;
	  if (accountHeads.containsKey(furnitureId)){
		  System.out.println("inside the iff Received furniture ID: "+furnitureId);

		  account =  (Furniture) furnitures.get(furnitureId);
		  
	  }
	  
	  account =  (Furniture) furnitures.get(furnitureId);

	  return account;
  }
  
  public Electronics getElectronics(String electronicsId){
	  System.out.println("Received Account ID: "+electronicsId);
	  Electronics electronic = null;
	  if (accountHeads.containsKey(electronicsId)){
		  System.out.println("inside the iff Received Account ID: "+electronicsId);

		  electronic =  (Electronics) electronics.get(electronicsId);
		  
	  }
	  electronic =  (Electronics) electronics.get(electronicsId);

	  return electronic;
  }
  
 
  
  @SuppressWarnings("unchecked")
public SelectResults<Item> queryItems() throws FunctionDomainException, TypeMismatchException, NameResolutionException, QueryInvocationTargetException{
	   QueryService queryService = cache.getQueryService();
	    Query query = null;
	    
	    // Query everything to show what's in the region
	    query = queryService.newQuery("SELECT DISTINCT * FROM /items");
	    System.out.println("\nExecuting query:\n\t" + query.getQueryString()); 
	    return (SelectResults<Item>) query.execute();
  }
  
  @SuppressWarnings("unchecked")
  public SelectResults<AccountHeads> queryAccountHeads() throws FunctionDomainException, TypeMismatchException, NameResolutionException, QueryInvocationTargetException{
	    System.out.println("\nExecuting query queryAccountHeads:\n\t"); 

	   QueryService queryService = cache.getQueryService();
	    Query query = null;
	    
	    // Query everything to show what's in the region
	    query = queryService.newQuery("SELECT DISTINCT * FROM /accountHeads");
	    System.out.println("\nExecuting query:\n\t" + query.getQueryString()); 
	    return (SelectResults<AccountHeads>) query.execute();
 }
  
  @SuppressWarnings("unchecked")
  public SelectResults<Item> queryRawMaterialItems() throws FunctionDomainException, TypeMismatchException, NameResolutionException, QueryInvocationTargetException{
	   QueryService queryService = cache.getQueryService();
	    Query query = null;
	    
	    // Query everything to show what's in the region
	    query = queryService.newQuery("  SELECT DISTINCT * FROM /items WHERE itemType = " + ItemType.RAW_MATERIALS.getCode() );
	    System.out.println("\nExecuting query:\n\t" + query.getQueryString()); 
	    return (SelectResults<Item>) query.execute();
 }
  
  @SuppressWarnings("unchecked")
  public SelectResults<PaymentVoucher> queryPaymentVouchers() throws FunctionDomainException, TypeMismatchException, NameResolutionException, QueryInvocationTargetException{
	   QueryService queryService = cache.getQueryService();
	    Query query = null;
	    
	    // Query everything to show what's in the region
	    query = queryService.newQuery("  SELECT DISTINCT * FROM /paymentVouchers" );
	    System.out.println("\nExecuting query:\n\t" + query.getQueryString()); 
	    return (SelectResults<PaymentVoucher>) query.execute();
}
  

 
  
 /* *//**
   * Get this list of all posts in the system
   * 
   * @return A list of the IDs of all of the posts in the system
 * @throws CqException 
 * @throws CqExistsException 
 * @throws QueryInvalidException 
 * @throws RegionNotFoundException 
 * @throws CqClosedException 
   *//*
  public Set<PostID> getPosts() {
    if(isClient) {
      return posts.keySetOnServer();
    } else {
      return posts.keySet();
    }
  }*/
  
  
	public void registerPaymentCQ() throws QueryInvalidException, CqExistsException, CqException, CqClosedException, RegionNotFoundException{
		// Get the query service
				QueryService queryService = cache.getQueryService();

				// Create CQ Attributes.
				CqAttributesFactory cqAf = new CqAttributesFactory();

				// Initialize and set CqListener.
				CqListener[] cqListeners = { new SimpleCqListener() };
				cqAf.initCqListeners(cqListeners);
				CqAttributes cqa = cqAf.create();

				// Construct a new CQ.
				String cqName = "AccountsPaybaleCQ";
				String query = "SELECT * FROM /paymentVouchers";
				
				System.out.println("Creating CQ \"" + cqName + "\" with query");
				System.out.println("\"" + query + "\"");

				// Execute the Cq. This registers the cq on the server.
				System.out.println("Executing CQ \"" + cqName + "\"...");
				CqQuery myCq = queryService.newCq(cqName, query, cqa);
				myCq.execute();
	}
	

	
 

  private void createItems(){
	  items.put(Item.ITEM_1_CODE, Item.createInstance(Item.ITEM_1));
	  items.put(Item.ITEM_2_CODE, Item.createInstance(Item.ITEM_2));

	  items.put(Item.ITEM_3_CODE, Item.createInstance(Item.ITEM_3));
	  items.put(Item.ITEM_4_CODE, Item.createInstance(Item.ITEM_4));
	  items.put(Item.ITEM_5_CODE, Item.createInstance(Item.ITEM_5));
  }
  
  
  private void createBooks(){
	  books.put(Book.BOOK_1_CODE, Book.createInstance(Book.BOOK_1));
	  books.put(Book.BOOK_2_CODE, Book.createInstance(Book.BOOK_2));
	  books.put(Book.BOOK_3_CODE, Book.createInstance(Book.BOOK_3));
	  books.put(Book.BOOK_4_CODE, Book.createInstance(Book.BOOK_4));
	  books.put(Book.BOOK_5_CODE, Book.createInstance(Book.BOOK_5));
  }
  
  private void createClothes(){
	  clothes.put(Clothing.CLOTHING_1_CODE, Clothing.createInstance(Clothing.CLOTHING_1));
	  clothes.put(Clothing.CLOTHING_2_CODE, Clothing.createInstance(Clothing.CLOTHING_2));
	  clothes.put(Clothing.CLOTHING_3_CODE, Clothing.createInstance(Clothing.CLOTHING_3));
	  clothes.put(Clothing.CLOTHING_4_CODE, Clothing.createInstance(Clothing.CLOTHING_4));
	  clothes.put(Clothing.CLOTHING_5_CODE, Clothing.createInstance(Clothing.CLOTHING_5));
  }
  
  private void createFurnitures(){
	  furnitures.put(Furniture.FURNITURE_1_CODE,Furniture.createInstance(Furniture.FURNITURE_1));
	  furnitures.put(Furniture.FURNITURE_2_CODE,Furniture.createInstance(Furniture.FURNITURE_2));
	  furnitures.put(Furniture.FURNITURE_3_CODE,Furniture.createInstance(Furniture.FURNITURE_3));
	  furnitures.put(Furniture.FURNITURE_4_CODE,Furniture.createInstance(Furniture.FURNITURE_4));
	  furnitures.put(Furniture.FURNITURE_5_CODE,Furniture.createInstance(Furniture.FURNITURE_5));
  }
  
  private void createDisks(){
	  disks.put(Disk.DISK_1_CODE, Disk.createInstance(Disk.DISK_1));
	  disks.put(Disk.DISK_2_CODE, Disk.createInstance(Disk.DISK_2));
	  disks.put(Disk.DISK_3_CODE, Disk.createInstance(Disk.DISK_3));
	  disks.put(Disk.DISK_4_CODE, Disk.createInstance(Disk.DISK_4));
	  disks.put(Disk.DISK_5_CODE, Disk.createInstance(Disk.DISK_5));
  }
  
  private void createElectronics(){
	  electronics.put(Electronics.ELECTRONICS_1_CODE, Electronics.createInstance(Electronics.ELECTRONICS_1));
	  electronics.put(Electronics.ELECTRONICS_2_CODE, Electronics.createInstance(Electronics.ELECTRONICS_2));
	  electronics.put(Electronics.ELECTRONICS_3_CODE, Electronics.createInstance(Electronics.ELECTRONICS_3));
	  electronics.put(Electronics.ELECTRONICS_4_CODE, Electronics.createInstance(Electronics.ELECTRONICS_4));
	  electronics.put(Electronics.ELECTRONICS_5_CODE, Electronics.createInstance(Electronics.ELECTRONICS_5));

  }
  
  private void createAccounts(){
	  accountHeads.put(AccountHeads.ACCOUNT_HEAD_1_CODE, AccountHeads.createInstance(AccountHeads.ACCOUNT_HEAD_1));
	  accountHeads.put(AccountHeads.ACCOUNT_HEAD_2_CODE, AccountHeads.createInstance(AccountHeads.ACCOUNT_HEAD_2));
	  accountHeads.put(AccountHeads.ACCOUNT_HEAD_3_CODE, AccountHeads.createInstance(AccountHeads.ACCOUNT_HEAD_3));
	  accountHeads.put(AccountHeads.ACCOUNT_HEAD_4_CODE, AccountHeads.createInstance(AccountHeads.ACCOUNT_HEAD_4));
	  
	  System.out.println("Account Head in the createAccounts method :"+getAccountHead(AccountHeads.ACCOUNT_HEAD_1_CODE));
  }
 
  private void createVendors(){
	  vendors.put(Vendor.VENDOR_1_CODE, Vendor.createInstance(Vendor.VENDOR_1));
	  vendors.put(Vendor.VENDOR_2_CODE, Vendor.createInstance(Vendor.VENDOR_2));
	  vendors.put(Vendor.VENDOR_3_CODE, Vendor.createInstance(Vendor.VENDOR_3));
	  vendors.put(Vendor.VENDOR_4_CODE, Vendor.createInstance(Vendor.VENDOR_4));

  }
  
 
  
  private static void writeToStdout(String msg) {
	    System.out.println(msg);
	  }
}
